package com.atguigu.day03;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Flink02_Transform_Map {
    public static void main(String[] args) throws Exception {
        //1.获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //2.从端口读取数据
//        DataStreamSource<String> streamSource = env.socketTextStream("localhost", 9999);
        DataStreamSource<String> streamSource = env.readTextFile("input/sensor.txt");

        //TODO 3.通过Map将读进来的数据转为WaterSensor
        SingleOutputStreamOperator<WaterSensor> result = streamSource.map(new MyMap());

        result.print();

        env.execute();
    }

//    public static class MyMap implements MapFunction<String, WaterSensor>{
    public static class MyMap extends RichMapFunction<String, WaterSensor> {
        //生命周期方法：会在方法的之前最先被调用，并且在每个并行实例执行一次  适用于创建连接代码
    @Override
    public void open(Configuration parameters) throws Exception {
        System.out.println("open");
    }

    //生命周期方法：会在方法的之前最后被调用，并且在每个并行实例执行一次  读文件时会调用两次 适用于关闭链接代码
    @Override
    public void close() throws Exception {
        System.out.println("close");
    }

    @Override
        public WaterSensor map(String value) throws Exception {
        System.out.println(getRuntimeContext().getTaskName());
        String[] split = value.split(",");
            return new WaterSensor(split[0],Long.parseLong(split[1]),Integer.parseInt(split[2]));
        }
    }
}
